package com.ybliang.plat.common.rpc.consumer.netty;

import com.ybliang.plat.common.rpc.core.RpcRequest;
import com.ybliang.plat.common.rpc.core.RpcResponse;
import com.ybliang.plat.common.rpc.protocol.RpcDecoder;
import com.ybliang.plat.common.rpc.protocol.RpcEncoder;
import com.ybliang.plat.common.rpc.protocol.json.JSON;
import com.ybliang.plat.common.rpc.protocol.json.JSONSerializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * @author liangyb1
 */
public class NettyClient {

    public static Logger LOGGER = LoggerFactory.getLogger(NettyClient.class);
    private static final EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    private static final int MAX_RETRY = 5;

    private String host;
    private Integer port;
    private Bootstrap bootstrap;
    private Channel channel;
    private NettyClientHandler nettyClientHandler;
    private ChannelFuture channelFuture;

    public NettyClient(String host, Integer port) {
        this.host = host;
        this.port = port;
    }

    public void connect() throws Exception {
        LOGGER.info("NettyClient->connect() start...");
        String delimiter = "_$";
        bootstrap = new Bootstrap();
        nettyClientHandler = new NettyClientHandler();
        try {
            bootstrap
                    .group(bossGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel channel) throws Exception {
                            //添加编码器
                            channel.pipeline().addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                            //添加解码器
                            channel.pipeline().addLast(new RpcDecoder(RpcResponse.class, new JSONSerializer()));
                            // 处理来自服务端的响应信息
                            channel.pipeline().addLast(nettyClientHandler);
                        }
                    });
//            connect(bootstrap, "127.0.0.1", 16789, 1);
            //连接到远程节点；等待连接完成
            channelFuture = bootstrap.connect(host, port).sync();
            LOGGER.info("NettyClient->connect() succeed...");
        }catch (Exception e) {
            LOGGER.error("NettyClient->connect() exception...", e);
        }

    }


    /**
     * 失败重连机制
     * @param bootstrap
     * @param host
     * @param port
     * @param retryTimes
     */
//    private void connect(Bootstrap bootstrap, String host, int port, final int retryTimes) throws InterruptedException {
//        ChannelFuture channelFuture = bootstrap.connect(host, port).sync().addListener(future -> {
//            if (future.isSuccess()) {
//                LOGGER.info("第{}次连接rpc服务端 成功....", retryTimes);
//            } else if (retryTimes >= MAX_RETRY) {
//                LOGGER.error("重试次数已用完，放弃连接");
//            } else {
//                int newRetryTimes = retryTimes + 1;
//                LOGGER.info("准备第{}次连接rpc服务端....", newRetryTimes);
//                bootstrap.config().group().schedule(() -> {
//                    try {
//                        connect(bootstrap, host, port, newRetryTimes);
//                    } catch (InterruptedException e) {
//                        e.printStackTrace();
//                    }
//                }, 5, TimeUnit.SECONDS);
//            }
//        });
//        channel = channelFuture.channel();
//    }


    /**
     * 发送
     * @param request
     * @return
     */
    public RpcResponse send(RpcRequest request) throws Exception {
//        channelFuture.channel().writeAndFlush(request);
        // 发送客户端的请求
        try {
            channelFuture.channel().writeAndFlush(request).await(3000);
        } catch (InterruptedException e) {
            LOGGER.error("NettyClient->send() request:{} exception...", JSON.toJSONString(request), e);
        }
        return nettyClientHandler.getRpcResponse(request.getId());
    }



}
